package com.learn.lb.kafka;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * @author laibo
 * @since 2019/8/26 14:59
 * ===================================
 * <p> kafka 消费者实例
 * ===================================
 */
public class KafkaConsumerDemo {

    public static void main(String[] args) {
        ConsumerConnector consumerConnector = createConn();
        Map<String, Integer> topicCountParam = new HashMap<>(2);
        topicCountParam.put(KafkaPropertiory.TOPIC, 1);
        Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumerConnector
                .createMessageStreams(topicCountParam);
        List<KafkaStream<byte[], byte[]>> kafkaStreams = messageStreams.get(KafkaPropertiory.TOPIC);
        if (null != kafkaStreams && kafkaStreams.size() > 0) {
            //因为上面只主题的配置为1
            KafkaStream<byte[], byte[]> messageAndMetadata = kafkaStreams.get(0);
            for (kafka.message.MessageAndMetadata<byte[], byte[]> kafkaMessage : messageAndMetadata) {
                byte[] message = kafkaMessage.message();
                System.out.println("***** 收到的消息：" + new String(message));
            }
        }

    }


    private static ConsumerConnector createConn() {
        Properties consumerProp = new Properties();
        consumerProp.put("bootstrap.servers", KafkaPropertiory.BROKER_LIST);
        consumerProp.put("zookeeper.connect", KafkaPropertiory.ZOOKEEPER_LIST);
        consumerProp.put("group.id", KafkaPropertiory.GEOUP_ID);
        ConsumerConfig consumerConfig = new ConsumerConfig(consumerProp);
        return Consumer.createJavaConsumerConnector(consumerConfig);
    }
}
